package com.weilive.core.netty.client;

import java.net.InetSocketAddress;
import java.net.URI;
import java.util.HashMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.http.HttpRequestEncoder;
import org.jboss.netty.handler.codec.http.HttpResponse;
import org.jboss.netty.handler.codec.http.HttpResponseDecoder;
import org.jboss.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
import org.jboss.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
import org.jboss.netty.handler.codec.http.websocketx.WebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.WebSocketVersion;

import com.weilive.core.protocol.json.JsonPacket;

public class NettyWebSocketClient {

	/********************************************************************************
	 * Client
	 ********************************************************************************/

	public static int TIMEOUT = 60* 1000;

	private final Channel channel;
	private final BlockingQueue<WebSocketFrame> frames;

	public NettyWebSocketClient(Channel channel, BlockingQueue<WebSocketFrame> frames) {
		this.channel = channel;
		this.frames = frames;
	}

	public JsonPacket doRequest(JsonPacket packet) throws Exception {
		if (channel.isConnected()) {
			channel.write(new TextWebSocketFrame(packet.encode()));
			WebSocketFrame frame = frames.poll(TIMEOUT, TimeUnit.MILLISECONDS);
			if (frame instanceof TextWebSocketFrame) {
				return JsonPacket.decode(((TextWebSocketFrame)frame).getText());
			}
		}
		return null;
	}

	public String doRequest(String request) throws Exception {
		if (channel.isConnected()) {
			channel.write(new TextWebSocketFrame(request));
			WebSocketFrame frame = frames.poll(TIMEOUT, TimeUnit.MILLISECONDS);
			if (frame instanceof TextWebSocketFrame) {
				return ((TextWebSocketFrame)frame).getText();
			}
		}
		return null;
	}

	public void asyncRequest(String request) throws Exception {
		channel.write(new TextWebSocketFrame(request));
	}

	public boolean isConnected() {
		return channel.isConnected();
	}

	public void close() {
		channel.close();
	}


	/********************************************************************************
	 * Manager
	 ********************************************************************************/

	public static class Manager {
		private final URI uri;
		private ClientBootstrap bootstrap;

		public Manager(String uri) throws Exception {
			this.uri = new URI(uri);
		}

		public void init() {
			if (bootstrap == null) {
				bootstrap = new ClientBootstrap(new NioClientSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool()));
				bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
					@Override
					public ChannelPipeline getPipeline() throws Exception {
						ChannelPipeline pipeline = Channels.pipeline();
						pipeline.addLast("decoder", new HttpResponseDecoder());
						pipeline.addLast("encoder", new HttpRequestEncoder());
						return pipeline;
					}
				});
			}
		}

		public void destory() {
			bootstrap.releaseExternalResources();
			bootstrap = null;
		}

		private WebSocketClientHandshaker newHandshaker() {
			HashMap<String, String> customHeaders = new HashMap<String, String>();
			customHeaders.put("MyHeader", "MyValue");
			// Connect with V13 (RFC 6455 aka HyBi-17). You can change it to V08 or V00.
			// If you change it to V00, ping is not supported and remember to change
			// HttpResponseDecoder to WebSocketHttpResponseDecoder in the pipeline.
			return  new WebSocketClientHandshakerFactory().newHandshaker(uri, WebSocketVersion.V13, null, false, customHeaders);
		}

		public NettyWebSocketClient createClient() {
			Channel channel = null;
			try {
				ChannelFuture future = bootstrap.connect(new InetSocketAddress(uri.getHost(), uri.getPort()));
				future.syncUninterruptibly();
				channel = future.getChannel();
				WebSocketClientHandshaker handshaker = newHandshaker();
				BlockingQueue<WebSocketFrame> frames = new LinkedBlockingQueue<WebSocketFrame>();

				channel.getPipeline().addLast("ws-handler", new Handler(handshaker, frames, channel));
				handshaker.handshake(channel).syncUninterruptibly();

				return new NettyWebSocketClient(channel, frames);
			} catch (Exception e) {
				if (channel != null) {
					channel.close();
					channel = null;
				}
			}
			return null;
		}
	}

	/********************************************************************************
	 * handler
	 ********************************************************************************/

	public static class Handler extends SimpleChannelUpstreamHandler {

		private final WebSocketClientHandshaker handshaker;
		private final BlockingQueue<WebSocketFrame> frames;

		public Handler(WebSocketClientHandshaker handshaker, BlockingQueue<WebSocketFrame> frames, Channel channel) {
			this.handshaker = handshaker;
			this.frames = frames;
		}

		@Override
		public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
			
		}

		@Override
		public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
		}

		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
			Channel ch = ctx.getChannel();
			if (!handshaker.isHandshakeComplete()) {
				handshaker.finishHandshake(ch, (HttpResponse) e.getMessage());
				return;
			}

			Object message = e.getMessage();
			if (message instanceof HttpResponse) {
			} else if (message instanceof PongWebSocketFrame) {
			} else if (message instanceof CloseWebSocketFrame) {
				ch.close();
			} else if (message instanceof PingWebSocketFrame) {
				ch.write(new PongWebSocketFrame(((PingWebSocketFrame)message).getBinaryData()));
			} else if (message instanceof WebSocketFrame) {
				frames.add((WebSocketFrame)message);
			}
		}

		@Override
		public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
			try {
				e.getCause().printStackTrace();
				e.getChannel().close();
			} catch (Exception ex) {
			}
		}
	}

}
